package Arcos::JobQueue;

use strict;
use warnings;

use Arcos::Log qw(debug critical);
use Arcos::DB::Job;
use Params::Validate qw(validate validate_pos SCALAR HASHREF);
use Config::ApacheFormat;
use File::Spec;
use Carp qw(croak);
use Arcos::Conf qw(ArcosRoot);

# config info cache
our $READ_CONFIG = 0;
our %TYPE_TO_CLASS;
our %MAX_PROCS_FOR_TYPE;

sub submit_job {
    my $class = shift;
    validate(@_,
             {type       => {type => SCALAR()},
              parameters => {type => HASHREF()},
             });
    my %params = @_;

    # check type for validity
    my @types = $class->available_types;
    croak(  "Invalid job type '$params{type}'.  "
          . "Check conf/queue.conf for legal values.")
      unless grep { $_ eq $params{type} } @types;

    my $job =
      Arcos::DB::Job->create({type       => $params{'type'},
                              parameters => $params{'parameters'}
                             });
    debug("job " . $job->id() . " added to queue");
    return $job->id();
}

sub accept_next_job {
    my $class = shift;
    my %args  = validate(@_, {types => 0});

    # append max_procs limits
    my @types;
    foreach my $type (@{$args{types}}) {
        my %type_spec = (type => $type);
        if ($MAX_PROCS_FOR_TYPE{$type}) {
            $type_spec{max_procs} = $MAX_PROCS_FOR_TYPE{$type};
        }
        push @types, \%type_spec;
    }

    my $job_id = Arcos::DB::Job->accept_next(\@types);
    if (defined $job_id) {
        debug("job " . $job_id . " accepted");
    }
    return $job_id;
}

sub is_complete {
    my $class = shift;
    validate_pos(@_, {type => SCALAR()});
    my $job_id = shift;
    return Arcos::DB::Job->is_complete($job_id);
}

sub get_result {
    my $class = shift;
    validate_pos(@_, {type => SCALAR()});
    my $job_id = shift;
    my $job    = Arcos::DB::Job->retrieve($job_id);
    return $job->result();
}

sub available_types {
    my $class = shift;
    $class->_read_config unless $READ_CONFIG;
    return keys %TYPE_TO_CLASS;
}

sub available_type_handlers {
    my $class = shift;
    $class->_read_config unless $READ_CONFIG;
    return %TYPE_TO_CLASS;
}

# read queue.conf
sub _read_config {
    my $class = shift;

    my $conf_file = File::Spec->catfile(ArcosRoot(), 'conf', 'queue.conf');
    unless (-e $conf_file and -r $conf_file) {
        croak "Unable to read configuration file '$conf_file': $!\n";
    }

    my $conf;
    eval {
        $conf = Config::ApacheFormat->new();
        $conf->read($conf_file);
    };
    if ($@ or !defined $conf) {
        die "Unable to read configuration file '$conf_file', error: $@\n";
    }

    # set up handler mapping
    foreach my $type_block ($conf->get('Type')) {
        my $job_type = $type_block->[1];
        my $block    = $conf->block(Type => $job_type);
        my $handler  = $block->get('Handler');
        $TYPE_TO_CLASS{$job_type} = $handler;
        debug(__PACKAGE__
              . " mapping $handler as handler for job type $job_type");
        $MAX_PROCS_FOR_TYPE{$job_type} = $block->get('MaxProcesses');
    }

    $READ_CONFIG = 1;
}

1;

__END__

=head1 NAME

Arcos::JobQueue - Interface to job queue system

=head1 SYNOPSIS

  # in client code
  my $job_id = Arcos::JobQueue->submit_job(
      type       => 'contribution',
      parameters => { contribution_id => 7 },
  );
  
  if ( Arcos::JobQueue->is_complete($job_id) ) {
      print "job is done\n";
      my $result = Arcos::JobQueue->get_result($job_id);
  }
  
  # in server code
  my $job_id = Arcos::JobQueue->accept_next_job();
  my $job_id = Arcos::JobQueue->accept_next_job( type => [ 'contribution' ]);
  my $job_id = Arcos::JobQueue->accept_next_job( type => [ 'contribution',
                                                           'mailing' ]);

  # get type information
  my @types     = Arcos::JobQueue->available_types;
  my %handlers  = Arcos::JobQueue->available_type_handlers;

=head1 DESCRIPTION

This is the Arcos job queue system.  It implements an asynchronous processing 
model with a simple FIFO queue stored in a database and the ability to decouple 
the job processing from the job submitter.  This is mostly useful for handling 
jobs that may take a while (credit card authorization) or that have the 
potential for getting flooded with requests during peak times.  The queue lets 
you put a limit on how many will be processing at once, and easily add more 
capacity to your system by putting in more servers to run the queue processing.

The job queue system is divided into multiple parts:

=over

=item arcos_queued

The daemon that handles watching for jobs and dispatching them.  This may be 
run on multiple servers.  Use C<arcos_queue_ctl> to start and stop it.

=item F<conf/queue.conf>

The configuration file that maps job types to handlers and sets basic 
parameters for the operation of the queue daemon.

=item Arcos::DB::Job

The low-level database interface to the queue data.  You should normally not 
need to use this directly.

=item Arcos::JobQueue

This class, the main interface for clients wanting to work with the job queue.

=item Arcos::JobQueue::Handler

This is the interface you need to implement to add a new job type to the queue.

=back

=head1 INTERFACE

=over 4

=item C<submit_job>

This is a class method, for submitting a job to be processed.  The job will be 
picked up by a queue server which may be on a different machine in the cluster.
It accepts the following parameters, both of which are required:

=over

=item C<type>

A string that matches one of the configured job types in F<conf/queue.conf>.

=item C<parameters>

A hash reference containing data to be passed to the job handler.  Everything 
in this hash ref MUST be serializable using Storable.  That means no database 
handles, open sockets, or strange XS objects.  The best approach is to stick 
with simple scalar data like the ID of an object to be fetched from a shared 
database.

=back

The return value from this method is an ID for the added job, which can be 
used to look it up with C<Arcos::DB::Job>.

=item C<accept_next_job>

This is a class method.  It returns the ID of the next pending job in the 
queue.  It can optionally be restricted to the next job of a specific 
type by passing the optional C<type> parameter:

=over

=item C<types>

An array-ref of strings that match one of the configured job types in
F<conf/queue.conf>.  If this parameter is not passed, any type of job
will be accepted.

=back

=item C<is_complete($job_id)>

Returns 1 if the specified job is completed or failed, or 0 otherwise.

=item C<get_result($job_id)>

Returns the result hash reference for the specified job.

=item C<available_types>

Class method which returns a list of legal values for the C<type>
parameter to C<accept_next_job> and C<submit_job>.

=item C<available_type_handlers>

Returns a hash of types mapped to handler classes, drawing data from
C<conf/queue.conf>.

=back

=head1 SEE ALSO

L<Arcos::DB::Job>, L<Arcos::JobQueue::Daemon>, L<Arcos::JobQueue::Daemon>
